package com.example.flinkcourse.lesson3;

import com.example.flinkcourse.lesson3.analytics.TransactionAnalytics;
import com.example.flinkcourse.lesson3.event.TransactionEventHandler;
import com.example.flinkcourse.lesson3.model.Transaction;
import com.example.flinkcourse.lesson3.sink.HBaseSinkFactory;
import com.example.flinkcourse.lesson3.source.KafkaSourceFactory;
import com.example.flinkcourse.lesson3.transform.TransactionProcessor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Flink Lesson 3 主作业类
 * 演示高级流处理功能
 */
public class FlinkLesson3Job {
    private static final Logger LOG = LoggerFactory.getLogger(FlinkLesson3Job.class);

    public static void main(String[] args) throws Exception {
        LOG.info("Starting Flink Lesson 3 Job...");
        
        // 创建执行环境
        Configuration configuration = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        
        // 配置检查点
        configureCheckpoint(env);
        
        // 配置重启策略
        configureRestartStrategy(env);
        
        // 创建 Kafka 数据源
        DataStream<Transaction> inputStream = env.fromSource(
            KafkaSourceFactory.createKafkaSource(),
            WatermarkStrategy.noWatermarks(),
            "Kafka Source"
        ).setParallelism(2);
        
        // 数据转换处理
        DataStream<Transaction> processedStream = TransactionProcessor.processTransactions(inputStream);
        
        // 流数据分析
        DataStream<TransactionStats> statsStream = TransactionAnalytics.analyzeTransactions(processedStream);
        
        // 事件驱动处理
        DataStream<Alert> alertStream = processedStream
            .keyBy(Transaction::getAccountId)
            .process(new TransactionEventHandler())
            .setParallelism(4);
        
        // 输出到 HBase
        processedStream.sinkTo(HBaseSinkFactory.createHBaseSink());
        
        // 执行作业
        LOG.info("Executing Flink Lesson 3 Job...");
        env.execute("Flink Lesson 3 Job");
    }
    
    /**
     * 配置检查点
     * @param env 流执行环境
     */
    private static void configureCheckpoint(StreamExecutionEnvironment env) {
        LOG.info("Configuring checkpoint...");
        
        // 启用检查点
        env.enableCheckpointing(60000);
        
        // 设置检查点模式
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        
        // 设置检查点超时
        env.getCheckpointConfig().setCheckpointTimeout(30000);
        
        // 设置检查点之间的最小暂停时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000);
        
        // 设置最大并发检查点数量
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        
        LOG.info("Checkpoint configuration completed");
    }
    
    /**
     * 配置重启策略
     * @param env 流执行环境
     */
    private static void configureRestartStrategy(StreamExecutionEnvironment env) {
        LOG.info("Configuring restart strategy...");
        
        // 设置固定延迟重启策略
        env.setRestartStrategy(
            RestartStrategies.fixedDelayRestart(
                3,  // 最大重启次数
                Time.seconds(10)  // 重启延迟
            )
        );
        
        LOG.info("Restart strategy configuration completed");
    }
} 